package system

import (
	"fmt"
	"github.com/astaxie/beego/logs"
	"github.com/eclipse/paho.mqtt.golang"
	"os"
	"os/signal"
	"smart_home/mqtt/conf"
	"syscall"
	"time"
)

// 订阅者
var subs []mqtt.Client

var (
	//MaxWorker = os.Getenv("MAX_WORKERS")
	//MaxQueue  = os.Getenv("MAX_QUEUE")
	MaxWorker = 100
	MaxQueue  = 5
)

// Job represents the job to be run
type Job struct {
	Client  mqtt.Client
	Message mqtt.Message
}

// A buffered channel that we can send work requests on.
var JobQueue chan Job

func init() {
	JobQueue = make(chan Job)
}

// Worker represents the worker that executes the job
type Worker struct {
	WorkerPool chan chan Job
	JobChannel chan Job
	quit       chan bool
}

func NewWorker(workerPool chan chan Job) Worker {
	return Worker{
		WorkerPool: workerPool,
		JobChannel: make(chan Job),
		quit:       make(chan bool)}
}

// Start method starts the run loop for the worker, listening for a quit channel in
// case we need to stop it
func (w Worker) Start() {
	go func() {
		for {
			// register the current worker into the worker queue.
			w.WorkerPool <- w.JobChannel

			select {
			case job := <-w.JobChannel:
				MatchRouter(job.Client, job.Message)

			case <-w.quit:
				// we have received a signal to stop
				return
			}
		}
	}()
}

// Stop signals the worker to stop listening for work requests.
func (w Worker) Stop() {
	go func() {
		w.quit <- true
	}()
}

type Dispatcher struct {
	// A pool of workers channels that are registered with the dispatcher
	WorkerPool chan chan Job
	maxWorkers int
}

func NewDispatcher(maxWorkers int) *Dispatcher {
	pool := make(chan chan Job, maxWorkers)
	return &Dispatcher{WorkerPool: pool, maxWorkers: maxWorkers}
}

func (d *Dispatcher) Run() {
	// starting n number of workers
	for i := 0; i < d.maxWorkers; i++ {
		worker := NewWorker(d.WorkerPool)
		worker.Start()
	}

	go d.dispatch()
}

func (d *Dispatcher) dispatch() {
	for {
		select {
		case job := <-JobQueue:
			// a job request has been received
			go func(job Job) {
				// try to obtain a worker job channel that is available.
				// this will block until a worker is idle
				jobChannel := <-d.WorkerPool

				// dispatch the job to the worker job channel
				jobChannel <- job
			}(job)
		}
	}
}

func Run() {
	logs.Debug("mqtt server runing....")
	done := make(chan os.Signal, 1)

	// 启动数据调度
	NewDispatcher(MaxWorker).Run()

	// 启动mqtt 并订阅数据
	subscriber()

	//signal.Notify(done)
	signal.Notify(done, os.Kill, os.Interrupt, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	logs.Debug("收到信号:", <-done)
	logs.Debug("开始退出...")
	logs.Debug("执行清理...")

	// 关闭订阅
	disconnect()

	logs.Debug("结束退出...")

	<-time.After(100)
	os.Exit(0)
}

// 订阅
func subscriber() {
	// 启动mqtt客户端
	for i := 0; i < conf.SubscriberNumber; i++ {
		client := connect(i)
		if client != nil {
			subs = append(subs, client)
		}
	}

	// 订阅主题
	for _, c := range subs {
		c.SubscribeMultiple(conf.Topics, func(client mqtt.Client, message mqtt.Message) {
			// let's create a job with the payload
			work := Job{Client: client, Message: message}
			// Push the work onto the queue.
			logs.Debug(message.Topic())
			JobQueue <- work
		})
	}
}

func connect(num int) mqtt.Client {
	cid := fmt.Sprintf("s_local_%d_%d", time.Now().UnixNano(), num)
	ops := mqtt.
		NewClientOptions().
		AddBroker(conf.Broker).
		SetUsername(conf.Username).
		SetPassword(conf.Password).
		SetClientID(cid).
		SetKeepAlive(time.Duration(conf.KeepAlive) * time.Second).
		SetPingTimeout(time.Duration(conf.PingTimeout) * time.Second).
		SetProtocolVersion(uint(conf.ProtocolVersion))

	client := mqtt.NewClient(ops)

	if token := client.Connect(); token.Wait() && token.Error() != nil {
		logs.Error(token.Error())
		logs.Info(cid, "connect fail")
		return nil
	}

	logs.Info(cid, "connect success")
	return client
}

func disconnect() {
	// 关闭mqtt
	for _, cli := range subs {
		cli.Disconnect(100)
	}
}
